#include "tcp_socket_connection.h"

#include "./../description_module/description_util.h"
#include "./../log_module/log_public_module.h"
#include "socket_address.h"
#include "socket_channel.h"
#include "socket_connection.h"
#include "socket_description.h"
#include "socket_errno.h"
#include "socket_util.h"

#include <cerrno>
#include <cstdio>
#include <mutex>

class TcpSocketChannel final : public SocketChannel
{
public:
    TcpSocketChannel(Loop * loop, const std::shared_ptr<Description> & description, TcpSocketConnection * connection);
    ~TcpSocketChannel() override;
    void OnErrorEvent(std::shared_ptr<SocketChannel> channel, std::shared_ptr<Description> description) override;
    void OnReadEvent(std::shared_ptr<SocketChannel> channel, std::shared_ptr<Description> description) override;
    void OnWriteEvent(std::shared_ptr<SocketChannel> channel, std::shared_ptr<Description> description) override;

private:
    TcpSocketConnection * m_connection;
};

TcpSocketChannel::TcpSocketChannel(
    Loop * loop,
    const std::shared_ptr<Description> & description,
    TcpSocketConnection * connection):SocketChannel(loop, description)
{
    m_connection = connection;
    LOG_SOCKET_MODULE("Create TcpSocketChannel(%p)", this);
}

TcpSocketChannel::~TcpSocketChannel()
{
    LOG_SOCKET_MODULE("Release TcpSocketChannel(%p)", this);
}

void TcpSocketChannel::OnErrorEvent(std::shared_ptr<SocketChannel> channel, std::shared_ptr<Description> description)
{
    LOG_SOCKET_MODULE("TcpSocketChannel(%p) OnErrorEvent description(%d) trigger error event", this, description->DescriptionData());
    SocketCommand command = {};
    command.SetCmd(SocketCommand::EmCmdError);
    command.SetConnection(m_connection->shared_from_this());
    m_connection->Update(&command);
    if (m_connection->ConnectState() == SocketConnection::EmConnectStateConnected)
    {
        m_connection->UpdateConnectState(SocketConnection::EmConnectStateDisconnected);
        SocketCommand dis_cmd = {};
        dis_cmd.SetCmd(SocketCommand::EmCmdDisconnect);
        dis_cmd.SetConnection(m_connection->shared_from_this());
        m_connection->Update(&dis_cmd);
    }
    channel->DelEvent();

}

void TcpSocketChannel::OnReadEvent(std::shared_ptr<SocketChannel> channel, std::shared_ptr<Description> description)
{
    LOG_SOCKET_MODULE("TcpSocketChannel(%p) OnReadEvent description(%d) trigger read event", this, description->DescriptionData());
    if (m_connection->ConnectState() != SocketConnection::EmConnectStateConnected)
    {
        return;
    }
    while (true)
    {
        if (m_connection->m_read_buffer.BufferFreeSize() < 65535)
        {
            m_connection->m_read_buffer.ExpendBuffer();
        }
        int result = SocketUtil::Read(description->DescriptionData(), m_connection->m_read_buffer);
        if (result != EmSocketErrno_SUCCESS)
        {
            LOG_SOCKET_MODULE("TcpSocketChannel(%p) OnReadEvent description(%d) on read event and errno:%d", this, description->DescriptionData(), errno);
            m_connection->UpdateConnectState(SocketConnection::EmConnectStateDisconnected);
            SocketCommand command = {};
            command.SetCmd(SocketCommand::EmCmdDisconnect);
            command.SetConnection(m_connection->shared_from_this());
            m_connection->Update(&command);
            m_connection->m_channel->DelEvent();
            break;
        }
        LOG_SOCKET_MODULE("TcpSocketChannel(%p) OnReadEvent description(%d) on read event and buffer length:%d", this, description->DescriptionData(), m_connection->m_read_buffer.BufferDataSize());
        if (m_connection->m_read_buffer.BufferFreeSize() > 0)
        {
            break;
        }
    }
    if (m_connection->ConnectState() == SocketConnection::EmConnectStateConnected)
    {
        if (m_connection->m_read_buffer.BufferDataSize() > 0)
        {
            SocketCommand command = {};
            command.SetCmd(SocketCommand::EmCmdRead);
            command.SetConnection(m_connection->shared_from_this());
            command.SetData(m_connection->m_read_buffer.BufferData(), m_connection->m_read_buffer.BufferDataSize());
            int length = m_connection->Update(&command) ? command.UseDataLength() : m_connection->m_read_buffer.BufferDataSize();
            m_connection->m_read_buffer.RemoveSize(length);
        }
        int event = EmIoIn | EmIoOneShot | EmIoRdHup;
        if (m_connection->m_write_buffer.BufferDataSize() > 0)
        {
            event |= EmIoOut;
        }
        channel->ModEvent(event);
    }
}

void TcpSocketChannel::OnWriteEvent(std::shared_ptr<SocketChannel> channel, std::shared_ptr<Description> description)
{
    LOG_SOCKET_MODULE("TcpSocketChannel(%p) OnWriteEvent description(%d) trigger write event", this, description->DescriptionData());
    if (m_connection->ConnectState() == SocketConnection::EmConnectStateConnecting)
    {
        SocketCommand command = {};
        command.SetConnection(m_connection->shared_from_this());
        int result = SocketUtil::Read(description->DescriptionData(), m_connection->m_read_buffer);
        if (result != EmSocketErrno_SUCCESS && result != EmSocketErrno_NEED_AGAIN)
        {
            command.SetCmd(SocketCommand::EmCmdError);
            m_connection->Update(&command);
            channel->DelEvent();
        }
        else
        {
            m_connection->UpdateConnectState(SocketConnection::EmConnectStateConnected);
            LOG_SOCKET_MODULE("TcpSocketChannel(%p) OnWriteEvent description(%d) update to connected state", this, description->DescriptionData());
            channel->ModEvent(EmIoRdHup | EmIoOneShot);
            command.SetCmd(SocketCommand::EmCmdConnect);
            m_connection->Update(&command);
        }
        if (m_connection->m_read_buffer.BufferDataSize() > 0)
        {
            SocketCommand read_cmd = {};
            read_cmd.SetConnection(m_connection->shared_from_this());
            read_cmd.SetCmd(SocketCommand::EmCmdRead);
            read_cmd.SetData(m_connection->m_read_buffer.BufferData(), m_connection->m_read_buffer.BufferDataSize());
            int length = m_connection->Update(&read_cmd) ? read_cmd.UseDataLength() : m_connection->m_read_buffer.BufferDataSize();
            m_connection->m_read_buffer.RemoveSize(length);
        }
    }
    else if (m_connection->ConnectState() == SocketConnection::EmConnectStateConnected)
    {
        std::unique_lock<std::mutex> lk(m_connection->m_mtx);
        if (m_connection->m_write_buffer.BufferDataSize() > 0)
        {
            SocketCommand command = {};
            command.SetConnection(m_connection->shared_from_this());
            auto result = SocketUtil::Write(description->DescriptionData(), m_connection->m_write_buffer);
            if (result != EmSocketErrno_SUCCESS)
            {
                m_connection->m_channel->DelEvent();
                m_connection->UpdateConnectState(SocketConnection::EmConnectStateDisconnected);
                command.SetCmd(SocketCommand::EmCmdDisconnect);
            }
            else
            {
                command.SetCmd(SocketCommand::EmCmdWrite);
            }
            m_connection->Update(&command);
        }
        if (m_connection->m_write_buffer.BufferDataSize() > 0)
        {
            channel->ModEvent(EmIoOut | EmIoOneShot | EmIoRdHup | EmIoIn);
        }
        else
        {
            channel->ModEvent(EmIoOneShot | EmIoRdHup | EmIoIn);
        }
    }
}

TcpSocketConnection::TcpSocketConnection(Loop * loop) : SocketConnection(EmConnectionTypeTcp)
{
    m_loop        = loop;
    m_channel     = nullptr;
    m_description = nullptr;
}

TcpSocketConnection::~TcpSocketConnection()
{
    LOG_SOCKET_MODULE("Release TcpSocketConnection(%p)", this);
}

int TcpSocketConnection::OpenSocket(SocketAddress address, Option option)
{
    m_description = std::make_shared<SocketDescription>();
    m_description->CreateDescription();
    m_channel = std::make_shared<TcpSocketChannel>(m_loop, m_description, this);
    return EmSocketErrno_SUCCESS;
}

int TcpSocketConnection::AsyncConnect(SocketAddress address, Option option)
{
    if (m_channel && ConnectState() != EmConnectStateConnected && ConnectState() == EmConnectStateInit)
    {
        auto self = shared_from_this();
        m_channel->Post([self, address, option]() {
            if (option.m_enable_nonblock)
            {
                DescriptionUtil::DescriptionBlockStatus(self->m_channel->DescriptionData(), 0);
            }
            SocketCommand command = {};
            command.SetConnection(self->shared_from_this());
            auto addr   = address.NetAddress();
            int  result = connect(self->m_channel->DescriptionData(), reinterpret_cast<struct ::sockaddr *>(&addr), sizeof(addr));
            if (result < 0 && errno != EINPROGRESS)
            {
                LOG_SOCKET_MODULE("TcpSocketConnection(%p) AsyncConnect description(%d) has error", self.get(), self->m_description->DescriptionData());
                self->UpdateConnectState(SocketConnection::EmConnectStateDisconnected);
                command.SetCmd(SocketCommand::EmCmdError);
                self->Update(&command);
            }
            else
            {
                if (result == 0)
                {
                    LOG_SOCKET_MODULE("TcpSocketConnection(%p) AsyncConnect description(%d) direct connected", self.get(), self->m_description->DescriptionData());

                    self->UpdateConnectState(EmConnectStateConnected);
                    self->m_channel->AddEvent(EmIoOneShot | EmIoRdHup);
                    command.SetCmd(SocketCommand::EmCmdConnect);
                    self->Update(&command);
                }
                else
                {
                    LOG_SOCKET_MODULE("TcpSocketConnection(%p) AsyncConnect description(%d) wait for write event", self.get(), self->m_description->DescriptionData());
                    self->UpdateConnectState(EmConnectStateConnecting);
                    self->m_channel->AddEvent(EmEvent::EmIoOut | EmEvent::EmIoOneShot);
                }
            }
        });
        return EmSocketErrno_SUCCESS;
    }
    return EmSocketErrno_CONNECT_FAILED;
}

int TcpSocketConnection::AsyncWrite(const uint8_t * data, int data_size)
{
    if (ConnectState() == EmConnectStateConnected)
    {
        {
            std::unique_lock<std::mutex> lk(m_mtx);
            m_write_buffer.PushDataToBuffer(data, data_size);
        }
        auto self = shared_from_this();
        m_channel->Post([self]() {
            if (self->m_channel->Index() > 0)
            {
                self->m_channel->ModEvent(EmIoOut | EmIoOneShot | EmIoRdHup | EmIoIn);
            }
            else
            {
                self->m_channel->AddEvent(EmIoOut | EmIoOneShot | EmIoRdHup | EmIoIn);
            }
        });
        return EmSocketErrno_SUCCESS;
    }
    return EmSocketErrno_SEND_FAILED;
}

int TcpSocketConnection::AsyncRead()
{
    if (ConnectState() == EmConnectStateConnected)
    {
        auto self = shared_from_this();
        m_channel->Post([self]() {
            int event = self->m_write_buffer.BufferDataSize() > 0 ? EmIoOut : 0;
            if (self->m_channel->Index() > 0)
            {
                self->m_channel->ModEvent(EmIoIn | EmIoOneShot | EmIoRdHup | event);
            }
            else
            {
                self->m_channel->AddEvent(EmIoIn | EmIoOneShot | EmIoRdHup | event);
            }
        });
        return EmSocketErrno_SUCCESS;
    }
    return EmSocketErrno_READ_FAILED;
}

void TcpSocketConnection::Close()
{
    UpdateConnectState(EmConnectStateDisconnected);
    if (m_channel)
    {
        auto self = shared_from_this();
        auto channel = m_channel;
        m_channel = nullptr;
        channel->Post([channel, self] {
            if (channel->Index() > 0)
            {
                channel->DelEvent();
            }
            if (self->m_description)
            {
                self->m_description->CloseDescription();
                self->m_description = nullptr;
            }
        });
    }
}
